'use strict';
const { Service } = require('egg');
const { Queue, Worker, QueueScheduler } = require('bullmq');
const fs = require('fs');
const path = require('path');

/**
 * BullMQ队列服务
 * 统一管理所有异步任务队列
 * @class QueueService
 * @extends {Service}
 */
class QueueService extends Service {
  constructor(ctx) {
    super(ctx);
    this.queues = new Map();
    this.workers = new Map();
    this.schedulers = new Map();
    this.initQueues();
  }

  /**
   * 初始化所有队列
   */
  initQueues() {
    const { bullmq } = this.config;
    if (!bullmq || !bullmq.queues || bullmq.queues.length === 0) {
      this.logger.warn('BullMQ队列配置为空，未初始化任何队列');
      return;
    }

    // 初始化每个队列
    bullmq.queues.forEach(queueConfig => {
      this.createQueue(queueConfig);
    });

    // 监听应用关闭事件，清理资源
    this.app.on('close', () => this.close());
  }

  /**
   * 创建单个队列及对应的worker
   * @param {Object} queueConfig - 队列配置
   * @param {string} queueConfig.name - 队列名称
   * @param {number} queueConfig.concurrency - 并发数
   */
  createQueue(queueConfig) {
    const { name, concurrency = 1 } = queueConfig;
    const { defaultQueueOptions } = this.config.bullmq;

    // 如果队列已存在，先关闭
    if (this.queues.has(name)) {
      this.logger.warn(`队列${name}已存在，将重新创建`);
      this.closeQueue(name);
    }

    // 创建队列实例
    const queue = new Queue(name, defaultQueueOptions);
    this.queues.set(name, queue);

    // 创建调度器，用于处理延迟任务和重试
    const scheduler = new QueueScheduler(name, { connection: defaultQueueOptions.connection });
    this.schedulers.set(name, scheduler);

    // 创建worker处理任务
    const worker = new Worker(name, async (job) => this.processJob(name, job), {
      connection: defaultQueueOptions.connection,
      concurrency
    });
    this.workers.set(name, worker);

    // 监听worker事件
    this._setupWorkerEvents(name, worker);

    this.logger.info(`队列${name}初始化完成，并发数: ${concurrency}`);
  }

  /**
   * 设置Worker事件监听
   * @param {string} queueName - 队列名称
   * @param {Worker} worker - Worker实例
   */
  _setupWorkerEvents(queueName, worker) {
    // 任务完成事件
    worker.on('completed', (job) => {
      this.logger.info(`队列${queueName}任务${job.id}完成，结果: ${JSON.stringify(job.returnvalue).substring(0, 100)}`);
    });

    // 任务失败事件
    worker.on('failed', (job, err) => {
      this.logger.error(`队列${queueName}任务${job.id}失败:`, err);
      // 记录失败任务到数据库
      this.ctx.service.failedJob.recordFailedJob(queueName, job, err);
    });

    // Worker错误事件
    worker.on('error', (err) => {
      this.logger.error(`队列${queueName}Worker错误:`, err);
    });
  }

  /**
   * 处理队列任务
   * @param {string} queueName - 队列名称
   * @param {Object} job - 任务对象
   * @returns {Promise<any>} 任务处理结果
   */
  async processJob(queueName, job) {
    this.logger.info(`开始处理队列${queueName}任务${job.id}: ${job.name}`);

    try {
      // 根据队列名称和任务名称路由到具体的处理函数
      const handlerPath = path.join(__dirname, 'queue', `${queueName}.js`);
      if (fs.existsSync(handlerPath)) {
        const handler = require(handlerPath);
        if (typeof handler[job.name] === 'function') {
          return await handler[job.name](this.ctx, job.data);
        }
        throw new Error(`队列${queueName}任务${job.name}处理函数不存在`);
      }
      throw new Error(`队列${queueName}处理文件不存在: ${handlerPath}`);
    } catch (error) {
      this.logger.error(`处理队列${queueName}任务${job.id}出错:`, error);
      throw error;
    }
  }

  /**
   * 添加任务到队列
   * @param {string} queueName - 队列名称
   * @param {string} jobName - 任务名称
   * @param {Object} data - 任务数据
   * @param {Object} options - 任务选项
   * @returns {Promise<Object>} 任务信息
   */
  async addJob(queueName, jobName, data, options = {}) {
    if (!this.queues.has(queueName)) {
      throw new Error(`队列${queueName}不存在`);
    }

    this.logger.info(`添加任务到队列${queueName}: ${jobName}`);
    const queue = this.queues.get(queueName);
    return await queue.add(jobName, data, options);
  }

  /**
   * 关闭单个队列
   * @param {string} queueName - 队列名称
   */
  async closeQueue(queueName) {
    if (this.workers.has(queueName)) {
      const worker = this.workers.get(queueName);
      await worker.close();
      this.workers.delete(queueName);
    }

    if (this.schedulers.has(queueName)) {
      const scheduler = this.schedulers.get(queueName);
      await scheduler.close();
      this.schedulers.delete(queueName);
    }

    if (this.queues.has(queueName)) {
      const queue = this.queues.get(queueName);
      await queue.close();
      this.queues.delete(queueName);
    }
  }

  /**
   * 关闭所有队列
   */
  async close() {
    this.logger.info('开始关闭所有BullMQ队列...');
    const queueNames = Array.from(this.queues.keys());
    for (const name of queueNames) {
      await this.closeQueue(name);
    }
    this.logger.info('所有BullMQ队列已关闭');
  }
}

module.exports = QueueService;